package etl;

import common.Constants;
import etl.util.LogParserUtil;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.Map;

public class ETL {
    public static class EtlToHdfsMapper extends Mapper<LongWritable, Text, LogWritable, NullWritable> {
        private static Logger logger = Logger.getLogger(ETL.EtlToHdfsMapper.class);
        private static int inputRecords, filterRecords, outputRecords;

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            inputRecords++;
            if (line.isEmpty()) {
                filterRecords++;
                return;
            }
            Map<String, String> buffer = LogParserUtil.LogParser(line);
            String eventName = buffer.get(Constants.LOG_EVENT_NAME);
            LogWritable k = new LogWritable();
            for (Map.Entry<String, String> en : buffer.entrySet()) {
                switch (en.getKey()) {
                    case "ver":
                        k.setVer(en.getValue());
                        break;
                    case "s_time":
                        k.setS_time(en.getValue());
                        break;
                    case "en":
                        k.setEn(en.getValue());
                        break;
                    case "u_ud":
                        k.setU_ud(en.getValue());
                        break;
                    case "u_mid":
                        k.setU_mid(en.getValue());
                        break;
                    case "u_sd":
                        k.setU_sd(en.getValue());
                        break;
                    case "c_time":
                        k.setC_time(en.getValue());
                        break;
                    case "l":
                        k.setL(en.getValue());
                        break;
                    case "b_iev":
                        k.setB_iev(en.getValue());
                        break;
                    case "b_rst":
                        k.setB_rst(en.getValue());
                        break;
                    case "p_url":
                        k.setP_url(en.getValue());
                        break;
                    case "p_ref":
                        k.setP_ref(en.getValue());
                        break;
                    case "tt":
                        k.setTt(en.getValue());
                        break;
                    case "pl":
                        k.setPl(en.getValue());
                        break;
                    case "ip":
                        k.setIp(en.getValue());
                        break;
                    case "oid":
                        k.setOid(en.getValue());
                        break;
                    case "on":
                        k.setOn(en.getValue());
                        break;
                    case "cua":
                        k.setCua(en.getValue());
                        break;
                    case "cut":
                        k.setCut(en.getValue());
                        break;
                    case "pt":
                        k.setPt(en.getValue());
                        break;
                    case "ca":
                        k.setCa(en.getValue());
                        break;
                    case "ac":
                        k.setAc(en.getValue());
                        break;
                    case "kv_":
                        k.setKv_(en.getValue());
                        break;
                    case "du":
                        k.setDu(en.getValue());
                        break;
                    case "browserName":
                        k.setBrowserName(en.getValue());
                        break;
                    case "browserVersion":
                        k.setBrowserVersion(en.getValue());
                        break;
                    case "osName":
                        k.setOsName(en.getValue());
                        break;
                    case "osVersion":
                        k.setOsVersion(en.getValue());
                        break;
                    case "country":
                        k.setCountry(en.getValue());
                        break;
                    case "province":
                        k.setProvince(en.getValue());
                        break;
                    case "city":
                        k.setCity(en.getValue());
                        break;
                }
            }
            try {
                outputRecords++;
                context.write(k,NullWritable.get());
            } catch (Exception e) {
                logger.error("etl输出异常",e);
            }
        }
    }

}
